//! 查询执行引擎模块
//!
//! 实现火山模型（Volcano Model）的查询执行引擎
//! 包括各种执行算子和执行上下文

use crate::{Error, Result};
use crate::storage::tuple::{Tuple, RowId};
use crate::storage::page::PageManager;
use crate::storage::{DataType, TableSchema}; // 使用存储引擎的类型
use crate::sql::Value;
use std::sync::Arc;
use std::collections::HashMap;

pub mod operators;
pub mod context;
pub mod batch;
pub mod operator_tests;
pub mod benchmarks;
pub mod integration_tests;

pub use operators::*;
pub use context::*;
pub use batch::*;

/// 执行算子特征
/// 火山模型的核心接口，每个算子都实现这个特征
pub trait Operator: Send + Sync {
    /// 打开算子，进行初始化
    fn open(&mut self, context: &mut ExecutionContext) -> Result<()>;
    
    /// 获取下一个元组
    /// 返回None表示没有更多数据
    fn next(&mut self, context: &mut ExecutionContext) -> Result<Option<ExecutionTuple>>;
    
    /// 关闭算子，清理资源
    fn close(&mut self, context: &mut ExecutionContext) -> Result<()>;
    
    /// 获取算子的输出模式（列信息）
    fn schema(&self) -> &ExecutionSchema;
    
    /// 获取算子的统计信息
    fn statistics(&self) -> OperatorStatistics;
    
    /// 重置算子状态，用于重新执行
    fn reset(&mut self) -> Result<()>;
}

/// 执行元组
/// 在执行过程中传递的数据单元
#[derive(Debug, Clone)]
pub struct ExecutionTuple {
    /// 元组数据
    pub values: Vec<Value>,
    /// 行ID（如果来自存储）
    pub row_id: Option<RowId>,
    /// 元组版本信息（用于MVCC）
    pub version_info: Option<VersionInfo>,
}

impl ExecutionTuple {
    /// 获取指定列的值
    pub fn get_value(&self, column_index: usize) -> Result<&Value> {
        self.values.get(column_index)
            .ok_or_else(|| Error::query_execution(format!("列索引 {} 超出范围", column_index)))
    }

    /// 获取列数量
    pub fn column_count(&self) -> usize {
        self.values.len()
    }
}

/// 执行模式
/// 描述算子输出的列信息
#[derive(Debug, Clone)]
pub struct ExecutionSchema {
    /// 列定义
    pub columns: Vec<ColumnInfo>,
}

/// 列信息
#[derive(Debug, Clone)]
pub struct ColumnInfo {
    /// 列名
    pub name: String,
    /// 数据类型
    pub data_type: DataType,
    /// 是否可为空
    pub nullable: bool,
    /// 表名（如果来自表）
    pub table_name: Option<String>,
}

// 移除重复的DataType定义，使用storage模块的DataType

/// 版本信息（用于MVCC）
#[derive(Debug, Clone)]
pub struct VersionInfo {
    /// 创建事务ID
    pub xmin: u32,
    /// 删除事务ID
    pub xmax: Option<u32>,
    /// 是否已提交
    pub committed: bool,
}

/// 算子统计信息
#[derive(Debug, Clone, Default)]
pub struct OperatorStatistics {
    /// 处理的元组数量
    pub tuples_processed: u64,
    /// 输出的元组数量
    pub tuples_output: u64,
    /// 执行时间（微秒）
    pub execution_time_us: u64,
    /// 内存使用量（字节）
    pub memory_usage_bytes: u64,
    /// 磁盘I/O次数
    pub disk_io_count: u64,
}

/// 执行上下文
/// 包含执行过程中需要的全局信息
#[derive(Debug)]
pub struct ExecutionContext {
    /// 页面管理器
    pub page_manager: Arc<PageManager>,
    /// 当前事务ID
    pub transaction_id: u32,
    /// 表元数据
    pub table_metadata: HashMap<String, TableMetadata>,
    /// 执行统计信息
    pub statistics: ExecutionStatistics,
    /// 内存限制（字节）
    pub memory_limit: usize,
    /// 当前内存使用量
    pub memory_usage: usize,
}

/// 表元数据
#[derive(Debug, Clone)]
pub struct TableMetadata {
    /// 表名
    pub name: String,
    /// 列信息
    pub columns: Vec<ColumnInfo>,
    /// 根页面ID
    pub root_page_id: Option<u64>,
    /// 行数
    pub row_count: u64,
    /// 表统计信息
    pub statistics: TableStatistics,
}

/// 表统计信息
#[derive(Debug, Clone, Default)]
pub struct TableStatistics {
    /// 行数估计
    pub estimated_row_count: u64,
    /// 页面数量
    pub page_count: u64,
    /// 平均行大小
    pub average_row_size: u32,
    /// 最后更新时间
    pub last_updated: u64,
}

/// 执行统计信息
#[derive(Debug, Default)]
pub struct ExecutionStatistics {
    /// 总执行时间
    pub total_execution_time_us: u64,
    /// 算子统计信息
    pub operator_stats: HashMap<String, OperatorStatistics>,
    /// 内存峰值使用量
    pub peak_memory_usage: usize,
    /// 总磁盘I/O
    pub total_disk_io: u64,
}

impl ExecutionTuple {
    /// 创建新的执行元组
    pub fn new(values: Vec<Value>) -> Self {
        Self {
            values,
            row_id: None,
            version_info: None,
        }
    }
    
    /// 创建带行ID的执行元组
    pub fn with_row_id(values: Vec<Value>, row_id: RowId) -> Self {
        Self {
            values,
            row_id: Some(row_id),
            version_info: None,
        }
    }
    
    /// 创建带版本信息的执行元组
    pub fn with_version(values: Vec<Value>, version_info: VersionInfo) -> Self {
        Self {
            values,
            row_id: None,
            version_info: Some(version_info),
        }
    }
    
    // get_value方法已在上面定义，移除重复定义
    
    /// 检查元组对当前事务是否可见
    pub fn is_visible_to_transaction(&self, transaction_id: u32) -> bool {
        if let Some(version_info) = &self.version_info {
            // 简化的可见性检查
            if version_info.xmin > transaction_id {
                return false;
            }
            
            if let Some(xmax) = version_info.xmax {
                if xmax <= transaction_id {
                    return false;
                }
            }
            
            version_info.committed
        } else {
            true // 没有版本信息，默认可见
        }
    }
}

impl ExecutionSchema {
    /// 创建新的执行模式
    pub fn new(columns: Vec<ColumnInfo>) -> Self {
        Self { columns }
    }
    
    /// 获取列数量
    pub fn column_count(&self) -> usize {
        self.columns.len()
    }
    
    /// 根据名称查找列索引
    pub fn find_column_index(&self, name: &str) -> Option<usize> {
        self.columns.iter().position(|col| col.name == name)
    }
    
    /// 获取列信息
    pub fn get_column(&self, index: usize) -> Option<&ColumnInfo> {
        self.columns.get(index)
    }
    
    /// 合并两个模式（用于JOIN）
    pub fn merge(&self, other: &ExecutionSchema) -> ExecutionSchema {
        let mut columns = self.columns.clone();
        columns.extend(other.columns.clone());
        ExecutionSchema::new(columns)
    }
    
    /// 投影模式（选择指定列）
    pub fn project(&self, column_indices: &[usize]) -> Result<ExecutionSchema> {
        let mut projected_columns = Vec::new();
        for &index in column_indices {
            if let Some(column) = self.columns.get(index) {
                projected_columns.push(column.clone());
            } else {
                return Err(Error::query_execution(format!("列索引 {} 超出范围", index)));
            }
        }
        Ok(ExecutionSchema::new(projected_columns))
    }
}

impl ColumnInfo {
    /// 创建新的列信息
    pub fn new(name: String, data_type: DataType, nullable: bool) -> Self {
        Self {
            name,
            data_type,
            nullable,
            table_name: None,
        }
    }
    
    /// 创建带表名的列信息
    pub fn with_table(name: String, data_type: DataType, nullable: bool, table_name: String) -> Self {
        Self {
            name,
            data_type,
            nullable,
            table_name: Some(table_name),
        }
    }
    
    /// 获取完全限定的列名
    pub fn qualified_name(&self) -> String {
        if let Some(table_name) = &self.table_name {
            format!("{}.{}", table_name, self.name)
        } else {
            self.name.clone()
        }
    }
}

// TableMetadata的实现已移动到context.rs

// ExecutionContext的实现已移动到context.rs

#[cfg(test)]
mod tests {
    use super::*;
    use crate::storage::page::PageManager;

    #[test]
    fn test_execution_tuple_creation() {
        let values = vec![
            Value::Integer(1),
            Value::Text("test".to_string()),
            Value::Boolean(true),
        ];

        let tuple = ExecutionTuple::new(values.clone());
        assert_eq!(tuple.values, values);
        assert!(tuple.row_id.is_none());
        assert!(tuple.version_info.is_none());
    }

    #[test]
    fn test_execution_tuple_with_version() {
        let values = vec![Value::Integer(42)];
        let version_info = VersionInfo {
            xmin: 100,
            xmax: None,
            committed: true,
        };

        let tuple = ExecutionTuple::with_version(values.clone(), version_info.clone());
        assert_eq!(tuple.values, values);
        assert_eq!(tuple.version_info.as_ref().unwrap().xmin, 100);
        assert!(tuple.version_info.as_ref().unwrap().xmax.is_none());
    }

    #[test]
    fn test_execution_tuple_visibility() {
        let values = vec![Value::Integer(1)];
        let version_info = VersionInfo {
            xmin: 100,
            xmax: Some(200),
            committed: true,
        };

        let tuple = ExecutionTuple::with_version(values, version_info);

        // 事务99看不到（创建事务未提交）
        assert!(!tuple.is_visible_to_transaction(99));

        // 事务150可以看到
        assert!(tuple.is_visible_to_transaction(150));

        // 事务200看不到（已被删除）
        assert!(!tuple.is_visible_to_transaction(200));

        // 事务250看不到（已被删除）
        assert!(!tuple.is_visible_to_transaction(250));
    }

    #[test]
    fn test_execution_schema() {
        let columns = vec![
            ColumnInfo::new("id".to_string(), DataType::Integer, false),
            ColumnInfo::new("name".to_string(), DataType::Text(Some(50)), true),
        ];

        let schema = ExecutionSchema::new(columns);
        assert_eq!(schema.column_count(), 2);
        assert_eq!(schema.find_column_index("id"), Some(0));
        assert_eq!(schema.find_column_index("name"), Some(1));
        assert_eq!(schema.find_column_index("age"), None);
    }

    #[test]
    fn test_execution_schema_project() {
        let columns = vec![
            ColumnInfo::new("id".to_string(), DataType::Integer, false),
            ColumnInfo::new("name".to_string(), DataType::Text(Some(50)), true),
            ColumnInfo::new("age".to_string(), DataType::Integer, true),
        ];

        let schema = ExecutionSchema::new(columns);
        let projected = schema.project(&[0, 2]).unwrap();

        assert_eq!(projected.column_count(), 2);
        assert_eq!(projected.get_column(0).unwrap().name, "id");
        assert_eq!(projected.get_column(1).unwrap().name, "age");
    }

    #[test]
    fn test_execution_schema_merge() {
        let left_columns = vec![
            ColumnInfo::new("id".to_string(), DataType::Integer, false),
            ColumnInfo::new("name".to_string(), DataType::Text(Some(50)), true),
        ];

        let right_columns = vec![
            ColumnInfo::new("age".to_string(), DataType::Integer, true),
            ColumnInfo::new("city".to_string(), DataType::Text(Some(100)), true),
        ];

        let left_schema = ExecutionSchema::new(left_columns);
        let right_schema = ExecutionSchema::new(right_columns);
        let merged = left_schema.merge(&right_schema);

        assert_eq!(merged.column_count(), 4);
        assert_eq!(merged.get_column(0).unwrap().name, "id");
        assert_eq!(merged.get_column(1).unwrap().name, "name");
        assert_eq!(merged.get_column(2).unwrap().name, "age");
        assert_eq!(merged.get_column(3).unwrap().name, "city");
    }

    #[test]
    fn test_column_info() {
        let column = ColumnInfo::new("test_col".to_string(), DataType::Text(Some(50)), true);
        assert_eq!(column.name, "test_col");
        assert_eq!(column.data_type, DataType::Text(Some(50)));
        assert!(column.nullable);
        assert!(column.table_name.is_none());

        let qualified_column = ColumnInfo::with_table(
            "col".to_string(),
            DataType::Integer,
            false,
            "table1".to_string(),
        );
        assert_eq!(qualified_column.qualified_name(), "table1.col");
    }

    #[test]
    fn test_execution_context() {
        let page_manager = Arc::new(PageManager::new(100));
        let mut context = ExecutionContext::new(page_manager, 123);

        assert_eq!(context.transaction_id, 123);
        assert_eq!(context.memory_usage, 0);
        assert_eq!(context.memory_limit, 64 * 1024 * 1024);

        // 测试内存分配
        context.allocate_memory(1024).unwrap();
        assert_eq!(context.memory_usage, 1024);

        // 测试内存释放
        context.deallocate_memory(512);
        assert_eq!(context.memory_usage, 512);

        // 测试内存限制检查
        context.set_memory_limit(1000);
        assert!(context.allocate_memory(600).is_err());
    }

    #[test]
    fn test_table_metadata() {
        let columns = vec![
            ColumnInfo::new("id".to_string(), DataType::Integer, false),
            ColumnInfo::new("name".to_string(), DataType::Text(Some(50)), true),
        ];
        let schema = ExecutionSchema::new(columns);

        let mut metadata = TableMetadata::new("users".to_string(), schema);
        assert_eq!(metadata.name, "users");
        assert_eq!(metadata.column_count(), 2);
        assert_eq!(metadata.find_column_index("id"), Some(0));
        assert!(metadata.root_page_id.is_none());

        metadata.set_root_page_id(42);
        assert_eq!(metadata.root_page_id, Some(42));
    }

    #[test]
    fn test_operator_statistics() {
        let mut stats = OperatorStatistics::default();
        assert_eq!(stats.tuples_processed, 0);
        assert_eq!(stats.tuples_output, 0);
        assert_eq!(stats.execution_time_us, 0);

        stats.tuples_processed = 100;
        stats.tuples_output = 80;
        stats.execution_time_us = 1000;

        assert_eq!(stats.tuples_processed, 100);
        assert_eq!(stats.tuples_output, 80);
        assert_eq!(stats.execution_time_us, 1000);
    }
}
